package org.apache.spark.examples.sql.streaming;

import org.apache.spark.examples.utils.Utils;
import org.apache.spark.sql.*;
import org.apache.spark.sql.expressions.Aggregator;

import java.io.Serializable;

/**
 * Created by admin on 2019/3/22.
 */
public class JavaUserDefinedTypedAggregation {

    public static final String filehdfsemployees = "hdfs://172.30.17.164:8020/streamsets/employees.json";

    public static void main(String[] args) {

        System.setProperty("hadoop.home.dir", "C:/hadoop-2.6.0");
        SparkSession spark = SparkSession
                .builder()
                .appName("Java Spark SQL user-defined Datasets aggregation example")
                .getOrCreate();

        Encoder<Employee> employeeEncoder = Encoders.bean(Employee.class);
        Dataset<Employee> ds = spark.read().json(Utils.filehdfsemployeesJson).as(employeeEncoder);
        ds.show();
        MyAverage myAverage = new MyAverage();
        //将函数转换为`TypedColumn`并为其命名
        TypedColumn<Employee,Double> averageSalary = myAverage.toColumn().name("average_salary");
        Dataset<Double> result = ds.select(averageSalary);
        result.show();
        spark.stop();
    }

    // $example on:typed_custom_aggregation$
    public static class Employee implements Serializable {
        private String name;
        private long salary;

        // Constructors, getters, setters...
        // $example off:typed_custom_aggregation$
        public String getName() {
            return name;
        }

        public void setName(String name) {
            this.name = name;
        }

        public long getSalary() {
            return salary;
        }

        public void setSalary(long salary) {
            this.salary = salary;
        }
        // $example on:typed_custom_aggregation$
    }

    public static class Average implements Serializable  {
        private long sum;
        private long count;

        // Constructors, getters, setters...
        // $example off:typed_custom_aggregation$
        public Average() {
        }

        public Average(long sum, long count) {
            this.sum = sum;
            this.count = count;
        }

        public long getSum() {
            return sum;
        }

        public void setSum(long sum) {
            this.sum = sum;
        }

        public long getCount() {
            return count;
        }

        public void setCount(long count) {
            this.count = count;
        }
        // $example on:typed_custom_aggregation$
    }

    /**
     * 类似于聚合操作
     */
    public static class MyAverage extends Aggregator<Employee, Average, Double> {
        //此聚合的零值。应满足任何b +零= b
        public Average zero() {
            return new Average(0L, 0L);
        }
        //组合两个值以生成新值。为了性能，该函数可以修改`buffer`
        //并返回它而不是构造一个新对象
        public Average reduce(Average buffer, Employee employee) {
            long newSum = buffer.getSum() + employee.getSalary();
            long newCount = buffer.getCount() + 1;
            buffer.setSum(newSum);
            buffer.setCount(newCount);
            return buffer;
        }
        // //合并两个中间值
        public Average merge(Average b1, Average b2) {
            long mergedSum = b1.getSum() + b2.getSum();
            long mergedCount = b1.getCount() + b2.getCount();
            b1.setSum(mergedSum);
            b1.setCount(mergedCount);
            return b1;
        }
        //转换缩小的输出
        public Double finish(Average reduction) {
            return ((double) reduction.getSum()) / reduction.getCount();
        }
        //指定中间值类型的
        public Encoder<Average> bufferEncoder() {
            return Encoders.bean(Average.class);
        }
        //指定最终输出值类型为
        public Encoder<Double> outputEncoder() {
            return Encoders.DOUBLE();
        }
    }
}
